package com.test.finkcdc.util;

import com.alibaba.fastjson.JSONObject;
import com.test.finkcdc.config.TargetConfig;
import com.test.finkcdc.entity.JobInfo;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.io.Serializable;
import java.lang.reflect.Field;
import java.sql.Connection;
import java.sql.PreparedStatement;

public class DeleteSink extends RichSinkFunction implements Serializable {
    private static final long serialVersionUID = -1786846316000464153L;
    private Connection conn;
    private PreparedStatement pre;
    private final JobInfo info;

    public DeleteSink(JobInfo info) {
        this.info = info;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        conn = TargetConfig.dataSource.getConnection();
        pre = conn.prepareStatement(info.getDeleteSql());
    }

    public void invoke(Object paramValue, Context context) throws Exception {
        JSONObject jsonObject = JSONObject.parseObject(paramValue + "");
        Object obj = JSONObject.parseObject(jsonObject.getString("jsonBean"), info.getEntity());
        Field field = obj.getClass().getDeclaredField(info.getTableId());
        field.setAccessible(true);
        Object value = field.get(obj);
        pre.setObject(1, value);
        pre.executeUpdate();
        conn.setAutoCommit(false);
        conn.commit();
    }

    public void close() throws Exception {
        super.close();
        if (pre != null) {
            pre.close();
        }
        if (conn != null) {
            conn.close();
        }
    }
}
